﻿
/************************************************************************/
/*                       对外通讯定义                                   */
/* |  4   |   2  |   1  |    4   | 长度-7 |                             */
/* +------+------+------+--------+--------+                             */
/* | 长度 | 类型 | 方向 | 序列号 | 数据区 |                             */
/* +------+------+------+--------+--------+                             */
/* 长度：整条通讯的长度，不包括<长度>本身                               */
/* 类型：整条通讯的操作命令                                             */
/* 方向：请求、响应或者无需响应的请求                                   */
/* 序列号：<方向>=响应，则序列号等于请求数据包的序列号                  */
/* 数据区：附加数据                                                     */
/*                                                                      */
/************************************************************************/
#include "stdafx.h"
#include "utlist.h"
#include "posix_socket.h"

static socket_comm_queue_t _in_comm; // 通讯输入队列
static socket_comm_queue_t _out_comm;// 通讯输出队列
static thrd_task_t _in_task; // 任务输入队列
static thrd_task_t _out_task;// 任务输出队列
static boost::thread *pthrd_task;//通讯线程指针
static boost::mutex _mutex;//队列互斥量
static boost::condition_variable _cond;//条件变量

static task_type_t typ_to_thrd[3] =
{
    TYP_LOG_GET,
    TYP_LOG_ENTRY,
    TYP_PROG_QUIT,
};

// Unified socket address. For IPv6 support, add IPv6 address structure
// in the union u.
union usa
{
    struct sockaddr sa;
    struct sockaddr_in sin;
#if defined(USE_IPV6)
    struct sockaddr_in6 sin6;
#endif
};

#define STRUCT_FILE_INITIALIZER {0, 0, 0, NULL, NULL}

// Describes listening socket, or socket which was accept()-ed by the master
// thread and queued for future handling by the worker thread.
typedef struct socket
{
    SOCKET sock;          // Listening socket
    union usa lsa;        // Local socket address
    union usa rsa;        // Remote socket address
} socket_t;

typedef uint32_t session_t; //用户ID数据类型

typedef struct client_info
{
    socket_t sock;
    uint8_t  data[COMM_DATA_MAX_SIZE * 2];
    uint32_t len;
    uint8_t  send_data[COMM_DATA_MAX_SIZE * 2];
    uint32_t send_len;
    session_t id;                        //ID
    bool     stop;
    struct client_info *next;
} client_info_t;



// 维护客户端的链表
static client_info_t *client_list_hdr = NULL;
static uint32_t client_id = 0;
static socket_t listening_sock;//服务器监听socket
static uint32_t listening_sock_num = 0;//服务器监听socket

static bool serv_flg = false;//关闭服务器标志
static bool stop_flg = false;//线程退出标志

/************************************************************************/
/* 维护客户端链表的函数                                                 */
/************************************************************************/


/************************************************************************/
/* @brief 读取客户端的通讯                                              */
/************************************************************************/
uint32_t client_read()
{
    client_info_t *ptr = NULL, *tmp = NULL;
    uint8_t buff[2048];
    int n, i, j, m = 0, len;
    socket_comm_entry_t comm;
    struct pollfd pfd[FD_SETSIZE];

    if(client_list_hdr == NULL)
    {
        return 0;
    }
    len = 0;
    LL_FOREACH_SAFE(client_list_hdr, ptr, tmp)
    {

        pfd[len].fd = ptr->sock.sock;
        pfd[len].events = POLLIN;
        len ++;
    }

    if (poll(pfd, len, 200) > 0)
    {
        for (i = 0; i < len; i++)
        {
            //处理输入事件
            if(pfd[i].revents == POLLIN)
            {
                LL_FOREACH_SAFE(client_list_hdr, ptr, tmp)
                {
                    //查找对应的FD
                    if((ptr->stop == false) && (ptr->sock.sock == pfd[i].fd))
                    {
                        n = recv(pfd[i].fd, (char *) & (ptr->data[ptr->len]), sizeof(ptr->data) - (ptr->len), 0);
                        if(n > 0)// 接收到了数据
                        {
                            ptr->len += n;
                            n = min(ptr->len, sizeof(ptr->data));
                            ptr->len = n;
                            // 以\r\n\r\n为通讯数据结束符
                            for(j = 0; j < n; j++)
                            {
                                // 搜索到第一个\r并且搜索到的位置小于最大长度-4
                                if((ptr->data[j] == '\r') && (j <= n - 4) )
                                {
                                    if((ptr->data[j + 1] == '\n')
                                            && (ptr->data[j + 2] == '\r')
                                            && (ptr->data[j + 3] == '\n'))
                                    {
                                        // 将收到的通讯放入输入队列中
                                        comm.data_size = j - 4;
                                        MEMCPY(comm.data, ptr->data, j);
                                        comm.remote_ip = inet_ntoa(ptr->sock.rsa.sin.sin_addr);
                                        comm.remote_port = ntohs(ptr->sock.rsa.sin.sin_port);
                                        comm.id = ptr->id;
                                        //boost::mutex::scoped_lock lock(_in_comm._mutex);
                                        _in_comm._queue.push(comm);
                                        //boost::mutex::scoped_lock lock1(_mutex);
                                        //_cond.notify_one();

                                        // 对接受缓冲区重新进行整理
                                        MEMCPY(buff, &ptr->data[j + 4], n - j - 4); //复制剩下的有效数据
                                        MEMCPY(ptr->data, buff, n - j - 4);
                                        ptr->len = n - j - 4;
                                        m ++;
                                        break;
                                    }
                                }
                            }
                        }
                        else
                        {
                            //连接已经关闭
                            ptr->stop = true;
                        }
                    }
                }

            }
        }
    }
    return m;
}
/************************************************************************/
/* @brief 向客户端发送数据                                              */
/************************************************************************/
uint32_t client_write()
{
    client_info_t *ptr = NULL, *tmp = NULL;
    int n, i, m = 0, len;
    socket_comm_entry_t comm;
    struct pollfd pfd[FD_SETSIZE];

    if(client_list_hdr == NULL)
    {
        return 0;
    }
    len = 0;
    LL_FOREACH_SAFE(client_list_hdr, ptr, tmp)
    {

        pfd[len].fd = ptr->sock.sock;
        pfd[len].events = POLLOUT;
        len ++;
    }

    if (poll(pfd, len, 200) > 0)
    {
        for (i = 0; i < len; i++)
        {
            //处理输入事件
            if(pfd[i].revents == POLLOUT)
            {
                LL_FOREACH_SAFE(client_list_hdr, ptr, tmp)
                {
                    //查找对应的FD
                    if((ptr->stop == false) && (ptr->sock.sock == pfd[i].fd))
                    {
                        n = (int)push(NULL, ptr->sock.sock, (const char *)ptr->send_data, ptr->send_len);
                        if(n > 0)
                        {
                            MEMSET(ptr->send_data, 0, ptr->send_len);
                            ptr->send_len = 0;
                        }
                        else
                        {
                            //连接已经关闭
                            //ptr->stop = true;
                            MEMSET(ptr->send_data, 0, ptr->send_len);
                            ptr->send_len = 0;
                        }
                    }
                }

            }
        }
    }
    return m;
}
/*
 *	@brief 设置socket设置超市参数
 */
static int set_sock_timeout(SOCKET sock, int milliseconds)
{
#ifdef _WIN32
    DWORD t = milliseconds;
#else
    struct timeval t;
    t.tv_sec = milliseconds / 1000;
    t.tv_usec = (milliseconds * 1000) % 1000000;
#endif
    return setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char *)(void *) &t, sizeof(t)) ||
           setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (const char *)(void *) &t, sizeof(t));
}

/*
 *	@brief 关闭连接
 */
static void close_socket_gracefully(client_info_t *pclient)
{
#if defined(_WIN32)
    char buf[8192];
    int n;
#endif
    struct linger linger;

    // Set linger option to avoid socket hanging out after close. This prevent
    // ephemeral port exhaust problem under high QPS.
    linger.l_onoff = 1;
    linger.l_linger = 1;
    setsockopt(pclient->sock.sock, SOL_SOCKET, SO_LINGER,
               (char *) &linger, sizeof(linger));

    // Send FIN to the client
    shutdown(pclient->sock.sock, SHUT_WR);
    set_non_blocking_mode(pclient->sock.sock);

#if defined(_WIN32)
    // Read and discard pending incoming data. If we do not do that and close the
    // socket, the data in the send buffer may be discarded. This
    // behaviour is seen on Windows, when client keeps sending data
    // when server decides to close the connection; then when client
    // does recv() it gets no data back.
    do
    {
        n = pull(NULL, pclient->sock.sock, buf, sizeof(buf));
    }
    while (n > 0);
#endif

    // Now we know that our FIN is ACK-ed, safe to close
    closesocket(pclient->sock.sock);
}

/*
 *	@brief 结束会话
 */
static void close_session(session_t id)
{
    client_info_t client;
    if(client_list_get(&client, id) == NULL)
    {
        return;
    }
    if (client.sock.sock != INVALID_SOCKET)
    {
        client.stop = true;
        close_socket_gracefully(&client);
        client_list_del(id);
    }
}

/************************************************************************/
/* @brief 接受客户端的链接                                              */
/************************************************************************/
static void accept_new_connection(const socket_t *listener)
{
    socket_t so;
    socklen_t len = sizeof(so.rsa);
    int on = 1;
    client_info_t info;
    thrd_task_entry_t task;

    if ((so.sock = accept(listener->sock, &so.rsa.sa, &len)) == INVALID_SOCKET)
    {
    }
    else
    {
        // 采集客户端信息
        getsockname(so.sock, &so.lsa.sa, &len);
        setsockopt(so.sock, SOL_SOCKET, SO_KEEPALIVE, (const char *)(void *) &on, sizeof(on));
        MEMSET(&info, 0, sizeof(info));
        MEMCPY(&info.sock, &so, sizeof(socket_t));
        info.stop = false;
        // 添加客户端
        client_list_add( &info);
        // 向所有线程广播有新的客户端
        task.src = ADDR_TCP;
        task.dst = ADDR_ALL;
        task.session = info.id;
        task.type = TYP_SERV_NEW_CLIENT;
        task.dir = DIR_NO_RESPONSE;
        task.seqno = 1;
        task.data_size = 0 ;
        send_task_entry(&task);
        // 记录日志
        uint8_t log[256];
        MEMSET(log, 0, sizeof(log));
        snprintf((char *)log, sizeof(log) - 1, "接受来自于%s:%u的传入连接，会话ID=%u",
                 inet_ntoa(so.rsa.sin.sin_addr), ntohs(so.rsa.sin.sin_port), info.id);
        logger_log(LOG_GRP_ALL, LOG_LVL_INFO, (const char *)log, strlen((char *)log));
    }
}

/************************************************************************/
/* @brief 监听客户端的链接                                              */
/************************************************************************/
static void listening(void)
{
    int on = 1, success = 1;

    if ((listening_sock.sock = socket(listening_sock.lsa.sa.sa_family, SOCK_STREAM, 6)) ==
            INVALID_SOCKET ||
            setsockopt(listening_sock.sock, SOL_SOCKET, SO_REUSEADDR,
                       (const char *)(void *) &on, sizeof(on)) != 0 ||
            bind(listening_sock.sock, &listening_sock.lsa.sa, sizeof(listening_sock.lsa)) != 0 ||
            listen(listening_sock.sock, SOMAXCONN) != 0)
    {
        closesocket(listening_sock.sock);
    }
    else
    {
        listening_sock_num = 1;
        char buf[512];
        snprintf(buf, sizeof(buf) - 1, "Listen socket on port = %d", ntohs(listening_sock.lsa.sin.sin_port));
        logger_log(LOG_GRP_ALL, LOG_LVL_INFO, buf, strlen(buf));
    }
}

/************************************************************************/
/* @brief tcp管理服务器通讯线程——处理任务队列                           */
/************************************************************************/
static void task_handler(thrd_task_entry_t &task)
{
    socket_comm_entry_t comm;
    socket_comm_type_t type;
    uint32_t n = 0;

    /// 处理输出通讯
    if(task.dst == ADDR_OUT)
    {
        unsigned char *ptr = comm.data;
        if(serv_flg)//
        {
            comm.id = task.session;
            comm.data_size = task.data_size + 7;
            // 长度
            ptr[n++] = comm.data_size >> 24;
            ptr[n++] = comm.data_size >> 16;
            ptr[n++] = comm.data_size >> 8;
            ptr[n++] = comm.data_size >> 0;
            // 类型
            type = type_from_thrd_to_comm(task.type);
            ptr[n++] = type >> 8;
            ptr[n++] = type >> 0;
            // 方向
            ptr[n++] = task.dir;
            // 序列号
            ptr[n++] = task.seqno >> 24;
            ptr[n++] = task.seqno >> 16;
            ptr[n++] = task.seqno >> 8;
            ptr[n++] = task.seqno >> 0;

            memcpy(&comm.data[n], task.data_buf, task.data_size);
            boost::mutex::scoped_lock lock(_out_comm._mutex);
            _out_comm._queue.push(comm);
            boost::mutex::scoped_lock lock1(_mutex);
            _cond.notify_one();
        }
        return;
    }
    switch(task.type)
    {
        /// 启动TCP管理服务器
    case TYP_SERV_START:
        switch(task.dir)
        {
        case DIR_REQUEST:
            break;
        case DIR_RESPONSE:
            break;
        case DIR_NO_RESPONSE:
#define DTYPE serv_data_t *
            DTYPE ptr = (DTYPE)(task.data_buf);
            if(ptr->port > 1024)//检查端口
            {
                listening_sock.lsa.sin.sin_addr.s_addr = htonl(0);
                listening_sock.lsa.sin.sin_family = AF_INET;
                listening_sock.lsa.sin.sin_port = htons((uint16_t) ptr->port);
                listening();
                serv_flg = true;
            }
#undef DTYPE
            break;
        }
        break;
        /// 停止线程
    case TYP_THRD_STOP:
        switch(task.dir)
        {
        case DIR_REQUEST:
            break;
        case DIR_RESPONSE:
            break;
        case DIR_NO_RESPONSE:
            stop_flg = true;
            break;
        }
        break;
    default:
        break;
    }
}

/************************************************************************/
/* @brief 输出通讯队列处理                                              */
/************************************************************************/
static void comm_output_handler(socket_comm_entry_t *pcomm)
{
    client_info_t *ptr, *tmp;
    int i = 0;

    if(client_list_hdr == NULL)
    {
        return ;
    }
    LL_FOREACH_SAFE(client_list_hdr, ptr, tmp)
    {
        // 连接没有关闭并且会话ID相等
        if((ptr->stop == false) && (ptr->id == pcomm->id))
        {
            // 剩余的空间不足以发送本次通讯内容
            if((sizeof(ptr->send_data) - ptr->send_len - 8) < pcomm->data_size)
            {

            }
            else
            {
                /*if(ptr->send_len < 4)
                {
                    ptr->send_len = 4;
                }*/
                // 数据内容
                MEMCPY(&ptr->send_data[ptr->send_len], pcomm->data, pcomm->data_size + 4);
                ptr->send_len += (pcomm->data_size + 4);
                // 数据结束
                MEMCPY(&ptr->send_data[ptr->send_len], "\r\n\r\n", 4);
                ptr->send_len += 4;
                // 数据长度
                //ptr->send_data[0] = ptr->send_len >> 24;
                //ptr->send_data[1] = ptr->send_len >> 16;
                //ptr->send_data[2] = ptr->send_len >> 8;
                //ptr->send_data[3] = ptr->send_len >> 0;
            }
        }
    }

}
/************************************************************************/
/* @brief 输入通讯队列处理                                              */
/************************************************************************/
static void comm_input_handler(socket_comm_entry_t &comm)
{
    thrd_task_entry_t task;
    socket_comm_type_t type;
    unsigned char *ptr;

    // 首先检查输入数据的长度
    if(comm.data_size < sizeof(task) - sizeof(task.data_buf))
    {
        ptr = comm.data;
        task.src = ADDR_TCP;
        task.dst = ADDR_IN;
        task.session = comm.id;
        type = socket_comm_type_t((ptr[4] << 8) + ptr[5]);
        task.type = type_from_comm_to_thrd(type);
        task.dir = (task_dir_t)ptr[6];
        task.seqno = ptr[7] << 24 |
                     ptr[8] << 16 |
                     ptr[9] << 8 |
                     ptr[10] << 0 ;
        task.data_size = comm.data_size - 7 ;
        MEMCPY(task.data_buf, &comm.data[11], task.data_size);
        // TODO-进行部分有效性的检查
        send_task_entry(&task);
        char buf[512];
        snprintf(buf, sizeof(buf) - 1, "Receive: session=%d, type=%d(%s), seqno=%d, data_size=%d",
            task.session, task.type, TASK_TYPE_TO_STRING(task.type), task.dir, task.seqno, task.data_size);
        logger_log(LOG_GRP_TCP, LOG_LVL_INFO, buf, strlen(buf));

    }
}
/************************************************************************/
/* @brief tcp管理服务器通讯线程                                         */
/************************************************************************/
static void tcp_comm_thrd_routin()
{
    thrd_task_entry_t task;
    socket_comm_entry_t comm;
    struct pollfd *pfd;
    int i, n;
    uint32_t wait_cnt = 3;
    client_info_t client;

    // 向交换线程注册
    register_thrd(NAME_TCP, &_in_task, NULL, NULL);
    pfd = (pollfd *)CALLOC(SERVER_LISTENING_NUM, sizeof(pfd[0]));
    /// 循环直到通知退出线程
    while(stop_flg == false)
    {
        wait_cnt = 3;
        //////////////////////////////////////////////////////////////////////////
        /// 处理任务队列
        while(stop_flg == false)
        {
            {
                // 查询是否有等待处理的任务
                boost::mutex::scoped_lock lock(_in_task._mutex);
                if(_in_task._queue.size() <= 0 )
                {
                    wait_cnt --;
                    break;
                }
                task = _in_task._queue.front();
                _in_task._queue.pop();
            }
            task_handler(task);
        }// 任务队列循环

        //////////////////////////////////////////////////////////////////////////
        /// 服务器正在运行
        if(serv_flg)
        {
            //io_service.poll();//让io_service处理已经准备好的事件
            /// 处理输出通讯队列
            while(stop_flg == false)
            {
                {
                    // 查询是否有等待处理的任务
                    boost::mutex::scoped_lock lock(_out_comm._mutex);
                    if(_out_comm._queue.size() <= 0 )
                    {
                        wait_cnt --;
                        break;
                    }
                    comm = _out_comm._queue.front();
                    _out_comm._queue.pop();
                }
                comm_output_handler(&comm);
            }

            /// 处理输入通讯队列
            while(stop_flg == false)
            {
                {
                    // 查询是否有等待处理的任务
                    boost::mutex::scoped_lock lock(_in_comm._mutex);
                    if(_in_comm._queue.size() <= 0 )
                    {
                        wait_cnt --;
                        break;
                    }
                    comm = _in_comm._queue.front();
                    _in_comm._queue.pop();
                }
                comm_input_handler(comm);
            }

            /// 处理poll

            for (i = 0; i < (int)listening_sock_num; i++)
            {
                pfd[i].fd = listening_sock.sock;
                pfd[i].events = POLLIN;
            }

            if (poll(pfd, listening_sock_num, 200) > 0)
            {
                for (i = 0; i < (int)listening_sock_num; i++)
                {
                    //处理输入事件
                    if(pfd[i].revents == POLLIN)
                    {
                        accept_new_connection(&listening_sock);
                    }
                }
            }

            client_write();
            /// 处理读取数据
            client_read();

            /// 处理要求关闭的客户端
            
            
            n = client_list_len();
            for(i=0; i<n; i++)
            {
                if(client_list_get(&client, i) != NULL)
                {
                    if(client.stop == true)
                    {
                        close_session(i);
                        // 向所有线程广播 
                        task.src = ADDR_TCP;
                        task.dst = ADDR_ALL;
                        task.session = i;
                        task.type = TYP_SERV_DEL_CLIENT;
                        task.dir = DIR_NO_RESPONSE;
                        task.seqno = 1;
                        task.data_size = 0 ;
                        send_task_entry(&task);
                        // 长度已经发生了变化
                        n = client_list_len();
                        i --;
                    }
                }
            }
        }// 服务器循环
        else
        {
            wait_cnt -= 2;
        }
        if(wait_cnt == 0)
        {
            //boost::mutex::scoped_lock lock1(_mutex);
            //_in_task._cond.wait(lock1);
            Sleep(100);
        }
    }// 主循环

    /// 推出线程之前的处理
    if(serv_flg == true)
    {
        logger_log(LOG_GRP_TCP, LOG_LVL_INFO, "Close TCP server.");
    }
    free(pfd);
    logger_log(LOG_GRP_TCP, LOG_LVL_WARNING, "Tcp server thread exit!");

}



/*
 *	@brief 启动一个TCP服务器
 */
int tcp_server_start()
{
    pthrd_task = new boost::thread(tcp_comm_thrd_routin);
    return 0;
}

/*
 *	@brief 停止服务线程
 */
void tcp_server_stop()
{
    delete pthrd_task;
}

/*
 *	@brief 启动服务器的端口监听功能
 */
void tcp_server_listen(int port)
{
    serv_data_t s;
    s.port = port;
    send_task(ADDR_TCP, ADDR_MAIN, 0, TYP_SERV_START, DIR_NO_RESPONSE, 0, sizeof(s), (unsigned char *)&s);
}

/************************************************************************/
/* @brief 将TYPE从线程的转换为对外的                                    */
/************************************************************************/
socket_comm_type_t type_from_thrd_to_comm(task_type_t type)
{
    socket_comm_type_t typ;
    int i;
    for(i = 0; i < sizeof(typ_to_thrd) / sizeof(typ_to_thrd[0]); i++)
    {
        if(typ_to_thrd[i] == type)
        {
            break;
        }
    }
    typ = (socket_comm_type_t)i;
    return typ;
}
/************************************************************************/
/* @brief 将TYPE从线程的转换为对外的                                    */
/************************************************************************/
task_type_t type_from_comm_to_thrd(socket_comm_type_t type)
{
    task_type_t typ;
    typ = typ_to_thrd[type];
    return typ;
}